package com.smsc.headend.task.engine.handler.impl.collection.strategy;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.core.map.MapUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.emeter.energyip.amiinterface.MeterReadsReplyMessageDocument;
import com.mchange.lang.LongUtils;
import com.smsc.headend.module.asset.entity.Meter;
import com.smsc.headend.module.data.dto.DataItemDTO;
import com.smsc.headend.module.data.dto.TaskDataDTO;
import com.smsc.headend.module.data.dto.TaskPacketDTO;
import com.smsc.headend.module.data.entity.RdMeterReadsLp;
import com.smsc.headend.module.data.enums.CollectionSource;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.*;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.manager.DdsFilterMdmDataManager;
import com.smsc.headend.task.engine.service.DataProcessService;
import com.smsc.headend.task.engine.service.FronzenDataService;
import com.smsc.headend.task.engine.service.collection.impl.CollectTaskServiceImpl;
import com.smsc.headend.task.engine.service.dds.DdsService;
import com.smsc.headend.task.engine.service.task.TaskService;
import com.smsc.headend.task.engine.utils.DdsXmlGenerateUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.poi.hssf.record.DVALRecord;
import org.assertj.core.util.Lists;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.CollectionUtils;

import java.nio.charset.Charset;
import java.util.*;
import java.util.stream.Collectors;

@Slf4j
public class ManualOnDemandReadStrategy implements CollectionStrategy {
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    TaskService taskService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    DataProcessService dataProcessService;
    @Autowired
    private DdsXmlGenerateUtil ddsXmlGenerateUtil;

    @Autowired
    DdsService ddsService;
    @Autowired
    FronzenDataService fronzenDataService;


    @Override
    public List<RdMeterReadsLp> filterCollectionData(Task task, AtomicTask atomicTask, Map<String, Object> param, List<RdMeterReadsLp> meterReadsCol) {
        return meterReadsCol;
    }

    private List<RdMeterReadsLp> doCompensate(List<RdMeterReadsLp> filteredRdMeterReadsLp, List<DataItemDTO> dataItemCosemList, Task task, Map<String, Object> param) {
        Boolean returnEmpty = false;
        Boolean compensateQueryOn = false;
        if (CollectionUtil.isEmpty(filteredRdMeterReadsLp)) {
            returnEmpty = true;
        }
        String compensateQuery = MapUtil.getStr(param, "compensateQuery");
        if ("daily".equals(compensateQuery)) {
            compensateQueryOn = true;
        }

        if (!returnEmpty || !compensateQueryOn) {
            return filteredRdMeterReadsLp;
        }
        Long lpDataItemId = Long.valueOf(MapUtil.getStr(param, "dataItemId"));
        //仅支持月曲线回空后的补偿计算
        if (!DdsFilterMdmDataManager.MONTH_LP.equals(lpDataItemId)) {
            return filteredRdMeterReadsLp;
        }
        log.info("startCompensateQueryDataForLpDataFetchEmpty:taskNo:{},meterId:{},param:{}", task.getTaskNo(), task.getMeterId(), task.getParamJson());
        long start = SystemClock.now();
        Long meterId = task.getMeterId();
        Long profileStartTv = Long.valueOf(MapUtil.getStr(param, CollectTaskServiceImpl.PROFILE_START_TIME));
        Long profileEndTv = Long.valueOf(MapUtil.getStr(param, CollectTaskServiceImpl.PROFILE_END_TIME));
        long endTv = (profileStartTv + profileEndTv) / 2;
        Date date = new Date(endTv * DateUtils.MILLIS_PER_SECOND);
        Date startDate = DateUtils.addMonths(date, -1);
        long startTv = startDate.getTime() / DateUtils.MILLIS_PER_SECOND;
        List<RdMeterReadsLp> dailyProfileData = dataProcessService.getRdMeterReadLpList(meterId, 1020070002L, null, startTv, endTv);
        if (CollectionUtil.isEmpty(dailyProfileData)) {
            log.info("noDailyDataExistsInThisMonth,taskNo:{},meterId:{},param:{}", task.getTaskNo(), task.getMeterId(), task.getParamJson());
            return filteredRdMeterReadsLp;
        }
        log.info("dailyDataExists,size:{}", dailyProfileData.size());
        boolean dailyMonthTvExists = dailyProfileData.stream().filter(p -> p.getTv().equals(endTv)).findAny().isPresent();
        List<RdMeterReadsLp> calculatedMonthlyData = null;

        if (dailyMonthTvExists) {
            calculatedMonthlyData = dailyProfileData.stream().filter(p -> p.getTv().equals(endTv)).collect(Collectors.toList());
        } else {
            calculatedMonthlyData = fronzenDataService.calculateMonthlyDataByMinMax(dailyProfileData, endTv);
        }
        calculatedMonthlyData = fronzenDataService.intersectWithCaptureObjList(calculatedMonthlyData, dataItemCosemList);
        List<String> dataItemIds = calculatedMonthlyData.stream().map(RdMeterReadsLp::getDataItemId).map(String::valueOf).collect(Collectors.toList());
        List<String> values = calculatedMonthlyData.stream().map(RdMeterReadsLp::getValue).collect(Collectors.toList());
        calculatedMonthlyData = fronzenDataService.handleSpecialDataItemData(calculatedMonthlyData);
        log.info("calculatedMonthData:msgId:{},meterId:{},dataItemId:{},value:{}", task.getMessageId(), task.getMeterId()
                , String.join(",", dataItemIds), String.join(",", values));
        return calculatedMonthlyData;
    }


    @Override
    public int callback(Meter meter, TaskReadData readData, Task task, AtomicTask atomicTask, Map<String, Object> param, List<DataItemDTO> dataItemCosemList, List<RdMeterReadsLp> filteredRdMeterReadsLp) {
        taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), readData.getProfileId().toString(), CollectionUtil.isEmpty(filteredRdMeterReadsLp) ? "no data" : JSONUtil.toJsonStr(filteredRdMeterReadsLp), Long.valueOf(1 * 60 * 60), true);
        if (StrUtil.equalsIgnoreCase(task.getTaskType(), TaskType.UAA_MeterReads.name())) {
            //MDM抄读
            if (isTaskLastAtomicTask(task, atomicTask)) {
                //所有原子任务完成后统一返回给adapter
                sendMdmDataToAdapter(task, meter);
            }
        }
        if (StrUtil.equalsIgnoreCase(task.getTaskType(), TaskType.Interface_ODR_Profile.name())) {
            //Nesco vending项目需求, 随抄月曲线数据后,采取补偿措施
            List<RdMeterReadsLp> compensateReadData = doCompensate(filteredRdMeterReadsLp, dataItemCosemList, task, param);
            List<TaskPacketDTO> taskPacketDTOList = taskService.getTaskPacket(task.getTaskNo());
            log.debug("Interface_ODR_Profile packet result are: {}", JSONUtil.toJsonStr(taskPacketDTOList));
            TaskResult taskResult = new TaskResult();
            taskResult.setMessageId(task.getMessageId());
            taskResult.setTaskNo(task.getTaskNo());
            taskResult.setResultData(CollectionUtils.isEmpty(compensateReadData) ? (CollUtil.isEmpty(readData.getTaskMeterReads()) ? null:readData.getTaskMeterReads()) : compensateReadData);

            Map<String, Object> dataItemMap = JSONUtil.toBean(atomicTask.getAtomicTaskParam(), Map.class);
            Map<String, Object> taskResultMap = JSONUtil.toBean(task.getParamJson(), Map.class);
            taskResultMap.putAll(dataItemMap);

            taskResultMap.put("packets", CollectionUtil.isEmpty(taskPacketDTOList) ? "" : taskPacketDTOList);
            taskResultMap.put("meterModelId", meter.getModelId());
            taskResultMap.put("meterCustomSn", meter.getCustomerSn());
            taskResultMap.put("profileDataItemId", readData.getProfileId().longValue());
            taskResult.setParamJson(JSONUtil.toJsonStr(taskResultMap));
            kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(Charset.forName("utf-8")));
        } else {
            log.error("odr data item reply without packets! taskNo: {}", task.getTaskNo());
        }
        if(StrUtil.equalsIgnoreCase(task.getTaskType(), TaskType.SwitchMeterPaymentPreToPost.name())) {
            if (CollectionUtil.isNotEmpty(filteredRdMeterReadsLp)) {
                ddsService.transformMeterReadResult(filteredRdMeterReadsLp);
            }

        }
        return 1;
    }

    private void sendMdmDataToAdapter(Task task, Meter meter) {
        List<RdMeterReadsLp> currentRdMeterReadsLp = new ArrayList<>();
        List<TaskDataDTO> taskDataDTOS = taskDAO.getAllTaskData(task.getTaskNo());
        for (TaskDataDTO taskDataDTO : taskDataDTOS) {
            Object value = taskDataDTO.getValue();
            String rdStr = (String) value;
            if (!"no data".equals(rdStr)) {
                List<RdMeterReadsLp> rdMeterReadsLps = JSONUtil.toList(rdStr, RdMeterReadsLp.class);
                currentRdMeterReadsLp.addAll(rdMeterReadsLps);
            }
        }
        TaskResult taskResult = new TaskResult();
        taskResult.setMessageId(task.getMessageId());
        taskResult.setTaskNo(task.getTaskNo());
        Map<String, Object> taskResultMap = JSONUtil.toBean(task.getParamJson(), Map.class);
        taskResultMap.put("meterModelId", meter.getModelId());
        taskResultMap.put("meterCustomSn", meter.getCustomerSn());
        MeterReadsReplyMessageDocument document = ddsXmlGenerateUtil.getDocumentLp(currentRdMeterReadsLp);
        taskResult.setDocumentData(document.toString());
        taskResult.setParamJson(JSONUtil.toJsonStr(taskResultMap));
        kafkaTemplate.send(TaskKafkaTopic.TASK_COMPLETE_RESULT_FROM_UTE_TO_ADAPTER, JSONUtil.toJsonStr(taskResult).getBytes(Charset.forName("utf-8")));
    }

    private boolean isTaskLastAtomicTask(Task task, AtomicTask atomicTask) {
        List<SubTaskGroup> subTaskGroups = task.getSubTaskGroups();
        if (subTaskGroups.size() == 0) {
            return true;
        }
        SubTaskGroup lastSubTaskGroup = subTaskGroups.get(subTaskGroups.size() - 1);
        List<AtomicTask> atomicTaskList = lastSubTaskGroup.getSubTasks();
        if (atomicTaskList.size() == 0) {
            return true;
        }
        AtomicTask lastAtomicTask = atomicTaskList.get(atomicTaskList.size() - 1);
        if (atomicTask.getAtomicTaskNo().equalsIgnoreCase(lastAtomicTask.getAtomicTaskNo())) {
            return true;
        }
        return false;
    }
}
